package org.mq.mymq.nameserver.event.listener;

import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.SlaveAckDTO;
import org.mq.mymq.common.enums.NameServerResponseCode;
import org.mq.mymq.common.event.Listener;
import org.mq.mymq.nameserver.common.CommonCache;
import org.mq.mymq.nameserver.event.model.SlaveReplicationMsgAckEvent;

/**
 * @ClassName SlaveReplicationMsgAckListener
 * @Description 主节点接受从节点同步ack信号处理器
 * @Author jiarong_ye
 * @Date 2025/6/23 18:16
 * @Version 1.0
 */
public class SlaveReplicationMsgAckListener implements Listener<SlaveReplicationMsgAckEvent> {
    @Override
    public void onReceive(SlaveReplicationMsgAckEvent event) throws Exception {
        String slaveAckMsgId = event.getMsgId();
        SlaveAckDTO slaveAckDTO = CommonCache.getAckMap().get(slaveAckMsgId);
        if (slaveAckDTO == null) {
            return;
        }
        Integer currentAckTime = slaveAckDTO.getNeedAckTime().decrementAndGet();
        // 如果是复制模式，代表所有从节点已经ack完毕
        // 如果是半同步复制模式，ack次数/2
        if (currentAckTime == 0) {
            CommonCache.getAckMap().remove(slaveAckMsgId);
            slaveAckDTO.getBrokerChannel().writeAndFlush(new TcpMsg(NameServerResponseCode.REGISTRY_SUCCESS.getCode(),
                    NameServerResponseCode.REGISTRY_SUCCESS.getDesc().getBytes()));
        }
//        throw new RuntimeException("error slave ack msg id: " + slaveAckMsgId);
    }
}
